package com.ysy.mq;


import com.alibaba.fastjson2.JSON;
import com.github.phantomthief.collection.BufferTrigger;
import com.ysy.constant.RedisConstant;
import com.ysy.constant.RocketMqConstant;
import com.ysy.entity.dto.NoteCollectDBDTO;
import com.ysy.entity.dto.NoteCollectDTO;
import com.ysy.entity.dto.NoteLikeOrUnlikeDTO;
import com.ysy.entity.dto.NoteLikeStatusCountDTO;
import com.ysy.enums.YesOrNoEnum;
import com.ysy.mapper.NoteCountMapper;
import com.ysy.persimmons.common.util.StringUtils;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 点赞集合消费
 * @author yangSy git地址：git地址：url:<a href="https://gitee.com/ysyGit01/persimmons"/>
 * @date 2025/5/2
 */

@Component
@RocketMQMessageListener(consumerGroup = "persimmons_count_group_note_collect",
                            topic = RocketMqConstant.NOTE_COLLECT_COUNT_TOPIC)
public class NoteCollectCountConsumer implements RocketMQListener<String> {


    private static Logger log = LoggerFactory.getLogger(NoteCollectCountConsumer.class);


    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(5000)
            .bufferSize(1000)
            .linger(Duration.ofSeconds(1))
            .setConsumerEx(this::BufferTriggerConsumerEx)
            .build();

    @Resource
    private NoteCountMapper noteCountMapper;

    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(String message) {
        log.info("NoteCollectCountConsumer start:{}",message);
        if (StringUtils.isBlank(message)) {return;}
        // 集合消息
        bufferTrigger.enqueue(message);
    }


    private void  BufferTriggerConsumerEx(List<String> messages) {
        log.info("NoteCollectCountConsumer BufferTriggerConsumerExSize:{}",messages.size());
        log.info("NoteCollectCountConsumer BufferTriggerConsumerEx:{}",messages);

        Map<Long, List<NoteCollectDTO>> collect = messages.stream()
                .map(json -> JSON.parseObject(json, NoteCollectDTO.class))
                .collect(Collectors.groupingBy(NoteCollectDTO::getUserId));

        List<NoteCollectDBDTO> countDTOList = new ArrayList<>();
        collect.forEach((userId, groupList) -> {
            int count = 0;
            for (NoteCollectDTO dto : groupList) {
                YesOrNoEnum yesOrNoEnum = YesOrNoEnum.get(dto.getType());
                if (Objects.isNull(yesOrNoEnum)) {
                    return; // 这里是跳过本次循环
                }
                switch (yesOrNoEnum) {
                    case YES -> count=count+1;
                    case NO ->  count=count-1;
                }
                NoteCollectDBDTO build = NoteCollectDBDTO.builder()
                        .userId(userId)
                        .noteId(dto.getNoteId())
                        .type(dto.getType())
                        .count(count)
                        .noteCreatorId(dto.getNoteCreatorId())
                        .build();
                countDTOList.add(build);
            }
        });

        // 执行redis存储和落库操作，害怕数据库顶不住可以用mq削峰或者限流
        for (NoteCollectDBDTO countDTO : countDTOList) {
            // 笔记维度统计，该笔记被收藏数
            String noteCountKey =RedisConstant.getCollectCountKey(countDTO.getNoteId());
            Boolean noteBool = redisTemplate.hasKey(noteCountKey);
            if (noteBool){
                // 更新笔记收藏数
                redisTemplate.opsForHash().increment(noteCountKey,RedisConstant.HASH_FIELD_COLLECT_TOTAL,countDTO.getCount());
            }
            // 用户维度统计,用户获赞和收藏总数
            String userCountKey =RedisConstant.getUserCountKey(countDTO.getNoteCreatorId());
            Boolean userBool = redisTemplate.hasKey(userCountKey);
            if (userBool){
                // 更新用户获点赞数
                redisTemplate.opsForHash().increment(userCountKey,RedisConstant.HASH_FIELD_COLLECT_TOTAL,countDTO.getCount());
            }
        }

        MessageBuilder<String> message = MessageBuilder.withPayload(JSON.toJSONString(countDTOList));
        rocketMQTemplate.asyncSend(RocketMqConstant.NOTE_COLLECT_COUNT_DB_TOPIC, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
            }

            @Override
            public void onException(Throwable throwable) {
            }
        });


    }


}
